package SparkStreamingKafka


import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.deser.std.StringDeserializer


/**
  * Kafka 0.10的Spark Streaming集成(spark获取kafka数据的最新方式)
  */
object KafkaDirectStream {

  def main(args: Array[String]): Unit = {
    //创建SparkConf，如果将任务提交到集群中，那么要去掉.setMaster("local[2]")

    val conf = new SparkConf().setAppName("KafkaDirectStream").setMaster("local[1]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //创建一个StreamingContext，其里面包含了一个SparkContext
    val streamingContext = new StreamingContext(sc, Seconds(5))

    //配置kafka的参数
    /**
      * Kafka服务监听端口
      * 指定kafka输出key的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
      * 指定kafka输出value的数据类型及编码格式（默认为字符串类型编码格式为uft-8）
      * 消费者ID，随意指定
      * 指定从latest(最新)还是smallest(最早)处开始读取数据
      * 如果true,consumer定期地往zookeeper写入每个分区的offset
      */
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.11.31:9092", //kafka机器IP：端口
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "test-consumer-group",
      "auto.offset.reset" -> "latest",
      "partition.assignment.strategy" -> "org.apache.kafka.clients.consumer.RangeAssignor",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    //要监听的Topic，可以同时监听多个
    val topics = Array("SparkKafka")

    //在Kafka中记录读取偏移量
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      //位置策略（可用的Executor上均匀分配分区）
      LocationStrategies.PreferConsistent,
      //消费策略（订阅固定的主题集合）
      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
    )

    //迭代DStream中的RDD(KafkaRDD)，将每一个时间点对应的RDD取出来
    stream.foreachRDD { rdd =>
      //获取该RDD对应的偏移量
      val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      //取出对应的数据
      rdd.foreach { line =>
        println(line.key() + " " + line.value())
      }

      //异步更新偏移量到kafka中
      // some time later, after outputs have completed
      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
    }
    streamingContext.start()
    streamingContext.awaitTermination()
  }
}
